package com.yunji.framework_template.common.redis;

import java.lang.annotation.Annotation;
import java.util.ArrayList;
import java.util.List;

import javax.annotation.Resource;

import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisCommands;
import redis.clients.jedis.Protocol;
import redis.clients.util.SafeEncoder;

/**
 * 
 * @author fenglibin
 *
 */
@Component
@Slf4j
public class RedisDistributedLock {

	@Resource
	private RedisTemplate<String, Object> redisTemplate;

	// 超时时间，以毫秒为单位
	private final long expire = 2000;

	public static final String UNLOCK_LUA;

	private LockStrategy lockStrategy = new LockStrategy() {
		@Override
		public Class<? extends Annotation> annotationType() {
			return LockStrategy.class;
		}

		@Override
		public int times() {
			return 25;
		}

		@Override
		public long sleepTime() {
			return 250;
		}

		@Override
		public long expires() {
			return 1000 * 10;
		}
	};

	static {
		// 要在Redis中使用Lua脚本的功能，必须在Redis中安装Lua脚本的插件支持
		StringBuilder sb = new StringBuilder();
		sb.append("if redis.call(\"get\",KEYS[1]) == ARGV[1] ");
		sb.append("then ");
		sb.append("    return redis.call(\"del\",KEYS[1]) ");
		sb.append("else ");
		sb.append("    return 0 ");
		sb.append("end ");
		UNLOCK_LUA = sb.toString();
	}

	/**
	 * 原理是从redis中获取到的lockKey的值是不是存在，如果不存在表示写入的是当前值，表示锁获取成功；
	 * 如果获取到的值存在，再判断是否已经超过了指定的期限，如果超过了指定的期限，则认为锁获取成功，否则认为锁获取失败； 默认超时时间为2秒 。
	 * 
	 * @param lockKey 用于获取锁定的key
	 * @return true表示获取到锁，false表示未获取到锁
	 */
	public boolean getLockByGetSet(String lockKey) {
		return getLockByGetSet(lockKey, expire);
	}

	/**
	 * 原理是从redis中获取到的lockKey的值是不是存在，如果不存在表示写入的是当前值，表示锁获取成功；
	 * 如果获取到的值存在，再判断是否已经超过了指定的期限，如果超过了指定的期限，则认为锁获取成功，否则认为锁获取失败；
	 * 
	 * @param lockKey 用于获取锁定的key
	 * @param expire  过期时间，以毫秒为单位
	 * @return true表示获取到锁，false表示未获取到锁
	 */
	public boolean getLockByGetSet(String lockKey, long expire) {
		long now = System.currentTimeMillis();
		// Redis的GetSet返回的值必须是字符串，否则会抛异常，因而将其转换为字符串
		String nowTime = String.valueOf(now);
		String oldTime = null;
		// 判断用于锁定的key是否已经被设置了值，如果被设置了值，则用于控制后续的处理逻辑不再进行
		if ((oldTime = (String) redisTemplate.opsForValue().getAndSet(lockKey, nowTime)) != null) {
			// 检查锁lockKey的值是不是超过了设定的时间，如2秒钟，没有超过则返回，不继续处理后续的任务；
			// 注：这个逻辑有个问题，就是客户端在2秒钟之内不停的重试，就永远不会进入到后面的处理环节。
			// 不过针对正常的业务请求这个是可以约定的，针对非正常的请求，被拦截也很正常，所以这个问题不是问题。
			if (now - Long.parseLong(oldTime) < expire) {
				return false;
			}
			return true;
		}
		return true;
	}

	/**
	 * 根据setnx的命令的特性，如果写入数据成功，可以用于当示锁获取成功，写入失败表示未获取到锁。 默认超时时间为2秒
	 * 只会尝试写入一次，写入失败或成功都返回结果，不会连接的尝试多次。
	 * 
	 * @param lockKey
	 * @return true表示获取到锁，false表示未获取到锁
	 */
	public boolean getLockBySetNX(String lockKey) {
		return getLockBySetNX(lockKey, expire);
	}

	/**
	 * 根据setnx的命令的特性，如果写入数据成功，可以用于当示锁获取成功，写入失败表示未获取到锁。
	 * 只会尝试写入一次，写入失败或成功都返回结果，不会连接的尝试多次。
	 * 
	 * @param lockKey
	 * @param expire  过期时间，以毫秒为单位
	 * @return true表示获取到锁，false表示未获取到锁
	 */
	public boolean getLockBySetNX(String lockKey, long expire) {
		RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
		JedisCommands commands = (JedisCommands) connection.getNativeConnection();
		boolean con = false;
		do {
			long now = System.currentTimeMillis();
			// Redis的GetSet返回的值必须是字符串，否则会抛异常，因而将其转换为字符串
			String nowTime = String.valueOf(now);
			con = false;
			// 返回1表示锁获取成功，返回0表示锁取失败
			String result = commands.set(lockKey, nowTime, "NX", "PX", expire);
			if ("1".equals(result)) {
				return true;
			} else {
				String oldTime = (String) redisTemplate.opsForValue().get(lockKey);
				if (null != oldTime) {
					// 检查锁lockKey的值是不是超过了设定的时间，如2秒钟，如果超过了则继续尝试获取锁，
					// 直到获取到锁，或者数据未超期时退出，循环判断可以解决死锁的问题
					if (now - Long.parseLong(oldTime) >= expire) {// 数据已经过期了
						con = true;
					}
				}
			}
		} while (con);
		return false;
	}

	/**
	 * 通过Lua(要在Redis中使用Lua脚本的功能，必须在Redis中安装Lua脚本的插件支持)脚本从Redis中删除指定的Key，
	 * 满足删除的Key的值Value必须与传入的value是相等的才可以删除，
	 * 应用场景为多线程的环境下只允许删除自己写入的值的key，以避免删除其它线程写入的key。
	 * 
	 * @param key   Redis的Key
	 * @param value 用于唯一确定当前线程的值
	 * @return 是否删除成功
	 */
	public boolean releaseLock(String key, String value) {
		// 释放锁的时候，有可能因为持锁之后方法执行时间大于锁的有效期，此时有可能已经被另外一个线程持有锁，所以不能直接删除
		try {
			List<String> keys = new ArrayList<>();
			keys.add(key);
			List<String> args = new ArrayList<>();
			args.add(value);

			// 使用lua脚本删除redis中匹配value的key，可以避免由于方法执行时间过长而redis锁自动过期失效的时候误删其他线程的锁
			// spring自带的执行脚本方法中，集群模式直接抛出不支持执行脚本的异常，所以只能拿到原redis的connection来执行脚本
			RedisCallback<Long> callback = (connection) -> {
				Object nativeConnection = connection.getNativeConnection();
				// 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
				// 集群模式
				if (nativeConnection instanceof JedisCluster) {
					return (Long) ((JedisCluster) nativeConnection).eval(UNLOCK_LUA, keys, args);
				}

				// 单机模式
				else if (nativeConnection instanceof Jedis) {
					return (Long) ((Jedis) nativeConnection).eval(UNLOCK_LUA, keys, args);
				}
				return 0L;
			};
			Long result = redisTemplate.execute(callback);

			return result != null && result > 0;
		} catch (Exception e) {
			log.error("release lock occured an exception", e);
		} finally {
			// 清除掉ThreadLocal中的数据，避免内存溢出
			// lockFlag.remove();
		}
		return false;
	}
	/**
	 * 使用setNX通过多次尝试获取锁定，如果在指定的时间之内多次尝试都不能够获取到锁，则退出
	 * @param lockKey
	 * @return
	 */
	public boolean lockTryManyTimes(String lockKey) {
		// 处理次数
		int times = lockStrategy.times();
		// 下次请求等侍时间
		long sleepTime = lockStrategy.sleepTime();
		// 有效时间
		final long expires = lockStrategy.expires() / 1000;
		// 最大请求等侍时间
		final long maxWait = 2 * 1000;

		final byte[] key = keySerializer(lockKey);
		int i = 0;
		while (times-- >= 0) {
			Object ok = redisTemplate.execute(new RedisCallback<Object>() {

				@Override
				public Object doInRedis(RedisConnection connection) throws DataAccessException {
					Object _ok = connection.execute(Protocol.Command.SET.name(), key, new byte[] { 0 },
							SafeEncoder.encode("EX"), Protocol.toByteArray(expires), SafeEncoder.encode("NX"));

					return _ok;
				}
			});
			if (ok != null) {
				return true;
			}
			long t = Math.min(maxWait, sleepTime * i);
			_await(t);
			i++;
		}

		return false;
	}

	private byte[] keySerializer(String key) {
		return StringSerializer.INSTANCE.serialize(key);
	}
	
	private void _await(long sleepTime) {
		try {
			Thread.sleep(sleepTime);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}
